package cn.dwj.hadoop.mr.flowsum;

import cn.dwj.hadoop.mr.wordcount.WCMapper;
import cn.dwj.hadoop.mr.wordcount.WCReduce;
import cn.dwj.hadoop.mr.wordcount.WCRunner;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * Created by Administrator on 2017/12/27 0027.
 */
public class FlowSunRunner extends Configured implements Tool{

    public static void main(String[] args) throws Exception {
        System.setProperty("hadoop.home.dir","E:/project/study/hadoop-2.6.5");
        String[] strArr = new String[2];
        strArr[0] = "E:/hadoop/flowsum/input/";
        strArr[1] = "E:/hadoop/flowsum/output/";
        int res = ToolRunner.run(new Configuration(), new FlowSunRunner(), strArr);
        System.exit(res);
    }

    public int run(String[] args) throws Exception {

        Configuration conf = new Configuration();
        Job flowSumJob = Job.getInstance(conf);
        //设置整个job所在的那些类在哪个jar包中
        flowSumJob.setJarByClass(FlowSunRunner.class);
        //本job使用的map和reduce的类
        flowSumJob.setMapperClass(FlowSumMapper.class);
        flowSumJob.setReducerClass(FlowSunReduce.class);

        //指定输出数据的类型，对mapper和reduce的类型都起作用
        flowSumJob.setOutputKeyClass(Text.class);
        flowSumJob.setOutputValueClass(FlowBean.class);

        FileInputFormat.setInputPaths(flowSumJob, new Path[]{new Path(args[0])});
        FileOutputFormat.setOutputPath(flowSumJob, new Path(args[1]));//指定处理结果输出数据存放在路径

        return flowSumJob.waitForCompletion(true)?0:1;//将job提交给集群运行过程要不要提示出来

    }
}
